package org.test.service;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler;

public class StateMaster extends CompositeService {

	private Dispatcher dispatcher;
	private String jobId;

	public StateMaster(String name, String jobId) {
		super(name);
		this.jobId = jobId;
	}

	@Override
	protected void serviceInit(Configuration conf) throws Exception {

		dispatcher = new AsyncDispatcher();
		dispatcher.register(JobEventType.class, new JobStateMachine(jobId,new JobEventHandler()));
		addService((Service) dispatcher);
		super.serviceInit(conf);
	}

	public Dispatcher getDispatcher() {
		return dispatcher;
	}

	private class JobEventHandler implements EventHandler<JobEvent> {
		
		public void handle(JobEvent event) {
			if (event.getType() == JobEventType.BEGIN) {
				System.out.println("Job begin:" + event.getJobId());
				dispatcher.getEventHandler().handle(
						new JobEvent(jobId, JobEventType.BEGIN));

			} else if (event.getType() == JobEventType.END) {
				System.out.println("Job end:" + event.getJobId());

				dispatcher.getEventHandler().handle(
						new JobEvent(jobId, JobEventType.END));

			}
		}

	}

}
